#include "messaging.h"
#include "trace.h"

#include <arpa/inet.h>
#include <string.h>
#include <errno.h>

using namespace std;

fd_subscriber::fd_subscriber(int fd)
    : fd_(fd)
{

}

fd_subscriber::~fd_subscriber()
{
    // leave closing the fd as the responsibility of the caller
}

void fd_subscriber::on_message(const DataMessage *data)
{
    uint32_t msg_sz = htonl(data->ByteSize());
    int total_sz = sizeof(uint32_t) + data->ByteSize();
    char *buf = new char[total_sz];
    memcpy(buf, &msg_sz, sizeof(uint32_t));
    if (!data->SerializeToArray(buf + sizeof(uint32_t), data->ByteSize())) {
        TERR("error occurred while serializing data");
    }
    int n, solong = 0;
    while (solong < total_sz) {
        n = write(fd_, buf + solong, total_sz - solong);
        if (n == -1) {
            TERR("fd_subscriber::on_message error: %s", strerror(errno));
            break;
        }
        solong += n;
    }
    delete [] buf;
}

messaging::messaging(const string &myid, const vector<string> &options)
    : id_(myid)
{
    transport * underlying = new xmpp_transport(myid);
    trans_ = new stransport_lite(underlying);
    if (!trans_->connect(options)) {
        throw string("failed to initialize transport");
    }
    trans_->add_listener(this);
    pthread_mutex_init(&m_, NULL);
}

messaging::~messaging()
{
    pthread_mutex_lock(&m_);
    map<string, pthread_mutex_t *>::iterator it, end;
    for (it = locks_.begin(), end = locks_.end(); it != end; ++it) {
        pthread_mutex_destroy(it->second);
        delete it->second;
    }
    pthread_mutex_unlock(&m_);
    locks_.clear();
    sub_.clear();
    pthread_mutex_destroy(&m_);
}

void messaging::deliver(DataMessage *msg)
{
    msg->set_src(id_);
    int to_size = msg->dest_size();
    if (to_size == 0) {
        // broadcast
        trans_->bcast(msg->SerializeAsString());
    } else if (to_size == 1) {
        // unicast
        trans_->ucast(msg->dest(0), msg->SerializeAsString());
    } else {
        // multicast (unimplemented)
        ;
    }
}

void messaging::subscribe(const string &subject, subscriber *s)
{
    pthread_mutex_t *lock = NULL;
    pthread_mutex_lock(&m_);
    if (locks_.find(subject) == locks_.end()) {
        lock = new pthread_mutex_t;
        pthread_mutex_init(lock, NULL);
        locks_[subject] = lock;
    } else {
        lock = locks_[subject];
    }
    pthread_mutex_unlock(&m_);
    if (lock) {
        pthread_mutex_lock(lock);
        sub_[subject].insert(s);
        pthread_mutex_unlock(lock);
    }
}

void messaging::unsubscribe(const string &subject, subscriber *s)
{
    pthread_mutex_lock(&m_);
    sub_[subject].erase(s);
    pthread_mutex_unlock(&m_);
}

string messaging::id() const
{
    return id_;
}

void messaging::on_data(const string &src, const string &data)
{
    // FIXME we should implement at least a refcnt-based memory pool to resolve
    // the memory leak here!
    // static int count = 0; // see how bad we are about memory leaking
    DataMessage msg;
    // count++;
    if (msg.ParseFromString(data)) {
        notify(&msg);
    }
    // if (count % 5 == 0)
        //TDEBUG("%d data message objects have been allocated", count);
}

void messaging::notify(const DataMessage *msg)
{
    pthread_mutex_t *lock = NULL;
    pthread_mutex_lock(&m_);
    if (locks_.find(msg->subject()) != locks_.end()) {
        lock = locks_[msg->subject()];
    }
    pthread_mutex_unlock(&m_);
    if (lock) {
        pthread_mutex_lock(lock);
        const set<subscriber *> &ss = sub_[msg->subject()];
        // to allow random access
        vector<subscriber *> ss_copy(ss.begin(), ss.end());
        pthread_mutex_unlock(lock);
        for (unsigned i = 0; i< ss_copy.size(); ++i) {
            subscriber *s = ss_copy[i];
            s->on_message(msg);
        }
    }
}

